package com.ytktt.study.week12.task11.config;

import com.ytktt.study.week12.task11.properties.MqttProperties;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

/**
 * 入站配置
 *
 * @author ytk
 */
@Slf4j
@Configuration
@IntegrationComponentScan
public class MqttInboundConfiguration {

    private final MessageHandler messageHandler;
    private final MqttProperties mqttProperties;

    public MqttInboundConfiguration(@Qualifier("mqttMessageInboundHandler") MessageHandler messageHandler, MqttProperties mqttProperties) {
        this.messageHandler = messageHandler;
        this.mqttProperties = mqttProperties;
    }

    @Bean(name = "mqttInputChannel")
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(this.mqttProperties.getServerUrl(),
                        this.mqttProperties.getClientId(), this.mqttProperties.getTopic());
        adapter.setCompletionTimeout(this.mqttProperties.getCompletionTimeout());
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(this.mqttProperties.getQos());
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return this.messageHandler;
    }
}
